package org.ngbx.demo.flink.dd.mokcer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.ngbx.demo.flink.dd.DeviceDataEvent;

import javax.annotation.Nullable;

public class DeviceDataEventSerialization {
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private final String topic;

    public DeviceDataEventSerialization(String topic) {
        this.topic = topic;
    }

    public ProducerRecord<String, byte[]> serialize(
            final DeviceDataEvent message, @Nullable final Long timestamp) {
        try {
            //if topic is null, default topic will be used
            return new ProducerRecord<>(topic, objectMapper.writeValueAsBytes(message));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Could not serialize record: " + message, e);
        }
    }
}
